package net.bwie.realtime.vehicle.dwd.job.one;

import com.alibaba.fastjson.JSON;
import net.bwie.realtime.utils.DorisUtil;
import net.bwie.realtime.utils.KafkaUtil;
import net.bwie.realtime.vehicle.dwd.bean.AverageSpeed;
import net.bwie.realtime.vehicle.dwd.bean.MonitorInfo;
import net.bwie.realtime.vehicle.dwd.function.SpeedWindowFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
/*todo
*   4 实时卡口拥堵情况监控
    卡口的实时拥堵情况，其实就是和通过卡口的车辆平均车速和通过的车辆的数量有关，
    为了统计实时的平均车速，我设定一个滑动窗口，窗口长度是为5分钟，滑动步长为1分钟。
    平均车速=当前窗口内通过车辆的车速之和 / 当前窗口内通过的车辆数量 ；设置水位线，允许数据最长迟到5秒。
    * */

/**
 * @author ASUS
 */
public class AverageSpeedMonitor {

    public static void main(String[] args) throws Exception {

        //1.执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(3000L);

        //2.读取数据源
        DataStream<String> carStream = KafkaUtil.consumerKafka(env, "dwd_car_monitor");
//      carStream.print("kafka");

        //3.数据处理
        DataStream<String> resultStream = handle(carStream);
        resultStream.print("result");

        //4.数据输出
        DorisUtil.saveToDoris(resultStream, "vehicle_realtime_report", "dws_average_speed_report");

        //5.执行
        env.execute("AverageSpeedMonitor");
    }

    private static DataStream<String> handle(DataStream<String> stream) {

        SingleOutputStreamOperator<MonitorInfo> mapStream = stream.map(new MapFunction<String, MonitorInfo>() {
            @Override
            public MonitorInfo map(String value) throws Exception {
                return JSON.parseObject(value, MonitorInfo.class);
            }
        });


        // 窗口 滑动窗口，窗口长度是为5分钟，滑动步长为1分钟
        WindowedStream<MonitorInfo, String, TimeWindow> windowStream = mapStream.keyBy(v -> v.getMonitorId())
                .window(SlidingProcessingTimeWindows.of(Time.seconds(5), Time.seconds(1)));

        //聚合计算
        SingleOutputStreamOperator<AverageSpeed> applyStream = windowStream.apply(new SpeedWindowFunction());


        SingleOutputStreamOperator<String> resultStream = applyStream.map(averageSpeed -> {
            return String.format(
                    "{\"window_start_time\":\"%s\",\"window_end_time\":\"%s\",\"cur_date\":\"%s\",\"monitor_id\":\"%s\",\"avg_speed\":\"%s\",\"car_count\":\"%s\"}",
                    averageSpeed.getStartTime() != null ? averageSpeed.getStartTime() : "",
                    averageSpeed.getEndTime() != null ? averageSpeed.getEndTime() : "",
                    averageSpeed.getStartTime() != null ? averageSpeed.getStartTime().substring(0, 10) : "",
                    averageSpeed.getMonitorId() != null ? averageSpeed.getMonitorId() : "",
                    averageSpeed.getAvgSpeed() != 0 ? averageSpeed.getAvgSpeed() : "",
                    averageSpeed.getCarCount() != 0 ? averageSpeed.getCarCount() : ""

            );
        });
        return resultStream;
    }
}
